package com.zebra.remoting.connection;

import com.zebra.remoting.config.ServerType;
import com.zebra.remoting.protocol.RemotingCommand;
import com.zebra.remoting.util.RemotingUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.AttributeKey;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * An abstraction of socket channel.
 * @author sk
 */
@Data
@ToString(of={"severType",  "serverId", "channel"})
@EqualsAndHashCode(of = {"serverId", "channel"})
public class Connection {
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultConnectionFactory.class);
    private ServerType severType;
    private int serverId;
    private Channel channel;

    /** Attribute key for connection */
//    public static final AttributeKey<Connection> CONNECTION = AttributeKey.valueOf("connection");

    private AtomicBoolean closed  = new AtomicBoolean(false);
    /** the reference count used for this connection. If equals 2, it means this connection has been referenced 2 times */
    private final AtomicInteger  referenceCount   = new AtomicInteger();

    public static final AttributeKey<Connection> CONNECTION = AttributeKey.valueOf("connection");


    /**
     * Constructor
     *
     * @param channel associated channel
     */
    public Connection(Channel channel) {
        this.channel = channel;
        this.channel.attr(CONNECTION).set(this);
    }

    /**
     * Constructor
     *
     * @param channel associated channel
     * @param url associated url
     */
    public Connection(Channel channel, ServerType url, int serverId) {
        this(channel);
        this.severType = url;
        this.serverId = serverId;
    }

    /**
     * to check whether the connection is fine to use
     *
     * @return true if connection is fine
     */
    public boolean isFine() {
        return this.channel != null && this.channel.isActive();
    }

    /**
     * increase the reference count
     */
    public void increaseRef() {
        this.referenceCount.getAndIncrement();
    }

    /**
     * decrease the reference count
     */
    public void decreaseRef() {
        this.referenceCount.getAndDecrement();
    }


    /**
     * Get the netty channel of the connection.
     *
     * @return Channel
     */
    public Channel getChannel() {
        return this.channel;
    }

    public static Connection getByChannel(Channel channel){
        return channel.attr(Connection.CONNECTION).get();
    }

    public void writeAndFlush(RemotingCommand command){
        this.channel.writeAndFlush(command);
    }

    /**
     * Close the connection.
     */
    public void close() {
        if (closed.compareAndSet(false, true)) {
            try {
                if (this.getChannel() != null) {
                    this.getChannel().close().addListener(new ChannelFutureListener() {

                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {
                            if (logger.isInfoEnabled()) {
                                logger.info("Close the connection to remote address={}, result={}, cause={}",
                                        RemotingUtil.parseRemoteAddress(Connection.this.getChannel()), future.isSuccess(), future.cause());
                            }
                        }

                    });
                }
            } catch (Exception e) {
                logger.warn("Exception caught when closing connection {}", RemotingUtil.parseRemoteAddress(Connection.this.getChannel()), e);
            }
        }
    }

}
